package com.heima.schedule.service.Impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.common.constants.ScheduleConstants;
import com.heima.common.redis.CacheService;
import com.heima.model.common.dtos.ResponseResult;
import com.heima.model.common.enums.AppHttpCodeEnum;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.TaskInfo;
import com.heima.model.schedule.pojos.TaskInfoLogs;
import com.heima.schedule.mapper.TaskInfoLogsMapper;
import com.heima.schedule.mapper.TaskInfoMapper;
import com.heima.schedule.service.ScheduleService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.stream.Collectors;

@Service
@Transactional
@Slf4j
public class ScheduleServiceImpl implements ScheduleService {
    @Autowired
    private TaskInfoMapper taskInfoMapper;
    @Autowired
    private TaskInfoLogsMapper taskInfoLogsMapper;
    @Autowired
    private CacheService cacheService;

    /**
     * 添加延迟任务
     *
     * @param task 延迟任务
     */
    @Override
    public ResponseResult<Long> addTask(Task task) {
        try {
            // 1. 添加任务到数据库
            boolean dbSuccess = addTaskToDb(task);
            if (!dbSuccess) {
                throw new RuntimeException("Failed to add task to database");
            }

            // 2. 添加任务到Redis
            try {
                addTaskToCache(task);
            } catch (Exception e) {
                // 缓存操作失败，可以记录日志
                log.error("Failed to add task to cache, taskId: {}", task.getTaskId(), e);
            }

            return ResponseResult.okResult(task.getTaskId());
        } catch (Exception e) {
            log.error("Failed to add task", e);
            throw e; // 或者返回特定的错误码/抛出特定异常
        }
    }

    /**
     * 取消任务
     *
     * @param taskId 延迟任务id
     */
    @Override
    public ResponseResult<Boolean> cancelTask(long taskId) {
        if (taskId <= 0) {
            log.warn("Invalid taskId: {}", taskId);
            return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID, "Invalid taskId");
        }

        try {
            // 1. 更新数据库任务状态
            Task task = updateDb(taskId, ScheduleConstants.CANCELLED);
            if (task == null) {
                log.warn("Task not found or update failed, taskId: {}", taskId);
                return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID, "Task not found or update failed");
            }

            // 2. 删除 Redis 缓存
            try {
                removeTaskFromCache(task);
            } catch (Exception e) {
                log.error("Failed to remove task from cache, taskId: {}", taskId, e);
            }

            return ResponseResult.okResult(true);
        } catch (Exception e) {
            log.error("Failed to cancel task, taskId: {}", taskId, e);
            throw new RuntimeException("Failed to cancel task", e);
        }
    }

    /**
     * 按照类型和优先级拉取任务
     */
    @Override
    public ResponseResult<Task> poll(int type, int priority) {
        Task task = null;
        try {
            String key = type + "_" + priority;
            String task_json = cacheService.lRightPop(ScheduleConstants.TOPIC + key);
            if (StringUtils.isNotBlank(task_json)) {
                task = JSON.parseObject(task_json, Task.class);
                //更新数据库信息
                updateDb(task.getTaskId(), ScheduleConstants.EXECUTED);
            }
        } catch (Exception e) {
            log.error("poll task exception");
        }

        return ResponseResult.okResult(task);
    }

    /**
     * 未来数据定时刷新
     */
    @Scheduled(cron = "0 */1 * * * ?")
    public void refresh() {

        String token = cacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);
        if (StringUtils.isNotBlank(token)) {
            log.info("未来数据定时刷新---定时任务");

            //获取所有未来数据的集合key
            Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");
            for (String futureKey : futureKeys) {//future_100_50

                //获取当前数据的key  topic
                String topicKey = ScheduleConstants.TOPIC + futureKey.split(ScheduleConstants.FUTURE)[1];

                //按照key和分值查询符合条件的数据
                Set<String> tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());

                //同步数据
                if (!tasks.isEmpty()) {
                    cacheService.refreshWithPipeline(futureKey, topicKey, tasks);
                    log.info("成功的将 " + futureKey + " 刷新到了 " + topicKey);
                }
            }
        }
    }

    /**
     * 从数据库重新加载任务数据到缓存（每5分钟执行一次）。
     *
     * <p>该方法会在以下情况触发：
     * <ol>
     *   <li>服务启动时（通过 {@code @PostConstruct} 初始化）</li>
     *   <li>后续每5分钟定时执行（通过 {@code @Scheduled} 定时任务）</li>
     *   <li>如果不使用 {@code @PostConstruct}, 服务启动时缓存为空, 会到第一个5分钟才执行定时任务</li>
     * </ol>
     *
     * <p>执行流程：
     * <ol>
     *   <li>清空现有缓存</li>
     *   <li>查询未来5分钟内需要执行的任务数据</li>
     *   <li>将任务数据转换为缓存模型并批量写入缓存</li>
     * </ol>
     *
     * <p>日志说明：
     * <ul>
     *   <li>成功时记录同步的任务数量及耗时</li>
     *   <li>失败时记录异常堆栈</li>
     *   <li>无数据时记录提示信息</li>
     * </ul>
     *
     * @since 1.0
     * @see Scheduled
     * @see PostConstruct
     *
     * @example 示例日志输出：
     * 开始从数据库同步数据到缓存...
     * 从数据库查询到15条待同步任务
     * 成功同步15条任务数据到缓存
     * 数据同步完成，耗时: 120ms
     */
    @Scheduled(cron = "0 */5 * * * ?")
    @PostConstruct
    public void reloadDataFromDB() {
        long startTime = System.currentTimeMillis();
        log.info("开始从数据库同步数据到缓存...");
        try {
            // 1. 清空缓存
            clearCache();
            log.debug("缓存清空完成");

            // 2. 计算查询时间范围（当前时间+5分钟）
            LocalDateTime now = LocalDateTime.now();
            LocalDateTime endTime = now.plusMinutes(5);

            // 3. 批量查询未来5分钟内的任务
            List<TaskInfo> allTasks = taskInfoMapper.selectList(
                    Wrappers.<TaskInfo>lambdaQuery()
                            .lt(TaskInfo::getExecuteTime,
                                    Date.from(endTime.atZone(ZoneId.systemDefault()).toInstant()))
            );

            if (CollectionUtils.isEmpty(allTasks)) {
                log.info("未找到需要同步的任务数据");
                return;
            }

            log.debug("从数据库查询到{}条待同步任务", allTasks.size());

            // 4. 批量转换
            List<Task> tasksToAdd = allTasks.stream()
                    .map(taskInfo -> {
                        Task task = new Task();
                        BeanUtils.copyProperties(taskInfo, task);
                        task.setExecuteTime(taskInfo.getExecuteTime().getTime());
                        return task;
                    })
                    .collect(Collectors.toList());

            // 5. 添加到缓存
            tasksToAdd.forEach(this::addTaskToCache);
            log.info("成功同步{}条任务数据到缓存", tasksToAdd.size());
        } catch (Exception e) {
            log.error("数据库数据同步到缓存失败", e);
        } finally {
            long duration = System.currentTimeMillis() - startTime;
            log.info("数据同步完成，耗时: {}ms", duration);
        }
    }

    // 删除缓存中未来数据集合和当前消费者队列的所有key
    private void clearCache() {
        Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");// future_
        Set<String> topicKeys = cacheService.scan(ScheduleConstants.TOPIC + "*");// topic_
        cacheService.delete(futureKeys);
        cacheService.delete(topicKeys);
    }

    /**
     * 删除redis中的任务数据
     *
     * @param task 延迟任务id
     */
    private void removeTaskFromCache(Task task) {

        String key = task.getTaskType() + "_" + task.getPriority();

        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            // 已到执行时间, 从list中删除
            cacheService.lRemove(ScheduleConstants.TOPIC + key, 0, JSON.toJSONString(task));
        } else {
            // 未到执行时间, 从zset中删除
            cacheService.zRemove(ScheduleConstants.FUTURE + key, JSON.toJSONString(task));
        }
    }

    /**
     * 删除任务，更新任务日志状态
     *
     * @param taskId 延迟任务id
     * @param status 状态
     */
    private Task updateDb(long taskId, int status) {
        Task task = null;
        try {
            //删除任务
            taskInfoMapper.deleteById(taskId);

            // 修改任务日志状态
            TaskInfoLogs taskInfoLogs = taskInfoLogsMapper.selectById(taskId);
            taskInfoLogs.setStatus(status);
            taskInfoLogsMapper.updateById(taskInfoLogs);

            // 转换为 Task 对象
            task = new Task();
            BeanUtils.copyProperties(taskInfoLogs, task);
            task.setExecuteTime(taskInfoLogs.getExecuteTime().getTime());

        } catch (Exception e) {
            log.error("task cancel exception taskid={}", taskId);
        }

        return task;
    }


    /**
     * 把任务添加到redis中
     *
     * @param task 延迟任务
     */
    private void addTaskToCache(Task task) {

        // 设置缓存的 key
        String key = task.getTaskType() + "_" + task.getPriority();

        //获取5分钟之后的时间  毫秒值
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE, 5);
        long nextScheduleTime = calendar.getTimeInMillis();

        //2.1 如果任务的执行时间小于等于当前时间，存入list
        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            cacheService.lLeftPush(
                    ScheduleConstants.TOPIC + key,
                    JSON.toJSONString(task));
            log.info("存入list：{}", task);
        } else if (task.getExecuteTime() <= nextScheduleTime) {
            //2.2 如果任务的执行时间大于当前时间 && 小于等于预设时间（未来5分钟） 存入zset中
            cacheService.zAdd(
                    ScheduleConstants.FUTURE + key, JSON.toJSONString(task),
                    task.getExecuteTime());
            log.info("存入zset：{}", task);
        }
    }

    /**
     * 添加任务到数据库中
     *
     * @param task 延迟任务
     */
    private boolean addTaskToDb(Task task) {

        boolean flag = false;

        try {
            //保存任务表
            TaskInfo taskInfo = new TaskInfo();
            BeanUtils.copyProperties(task, taskInfo);
            taskInfo.setExecuteTime(new Date(task.getExecuteTime()));
            taskInfoMapper.insert(taskInfo);

            //设置taskID
            task.setTaskId(taskInfo.getTaskId());

            //保存任务日志数据
            TaskInfoLogs taskinfoLogs = new TaskInfoLogs();
            BeanUtils.copyProperties(taskInfo, taskinfoLogs);
            taskinfoLogs.setVersion(1); // 乐观锁版本
            taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
            taskInfoLogsMapper.insert(taskinfoLogs);

            flag = true;
        } catch (Exception e) {
            log.error("延迟任务: {} 类型: {} 失败！. msg: ", task.getTaskType(), task.getTaskId(), e);
        }

        log.info("延迟任务: {} 类型: {} 添加成功！", task.getTaskId(), task.getTaskType());
        return flag;
    }
}